
#include "../include/fastArray.h"
#include  "../include/thread.h"
#include "../include/session.h"
#include "../include/mylog.h"

#include "ae.h"
#include "dict.h"

#include <assert.h>
#include <errno.h>
#include <stdarg.h>
#include <stdlib.h>
#include <stdio.h>
#include <signal.h>
#include <unistd.h>

static long long sThreadId = 0;

static void removeSessionEvent(struct session* session) {
	if (!session || !session->t) {
		return;
	}

	if (session->sendEventState == OPEN_SENDEVENT) {
		aeDeleteFileEvent(session->t->threadEventLoop, session->cfd, AE_WRITABLE);
		session->sendEventState = CLOSE_SENDEVENT;
	}
	aeDeleteFileEvent(session->t->threadEventLoop, session->cfd, AE_READABLE);
	return;
}

static void onFileReadEvent(struct aeEventLoop * eventLoop, int fd, void *clientData, int mask) {
	struct session* session = (struct session* )clientData;
	if (!session) {
		return;
	}

	int len;
	len = readSession(session);
	mylog(LOG_DEBUG, "on file read event, session id[%d], len[%d]", session->id, len);
	if (len <= 0) {
		if (session->callBackHandle) {
			if (session->callBackHandle->onDisConnectionCallBack) {
				session->callBackHandle->onDisConnectionCallBack(session, session->interData);
			}
		}

		//destroySession(session);
		destroySessionEvent(session->t, session);
		removeSessionEvent(session);
	} else if (len > 0){
		int ret;
		if (session->callBackHandle) {
			if (session->callBackHandle->onDataCallBack) {
				ret = session->callBackHandle->onDataCallBack(session, session->interData);
			}
		}

		mylog(LOG_INFO, "session id[%d], onDataCallBack ret[%d]", session->id, ret);
		if (ret == SHUTDOWN_HANDLE) {
			if (session->callBackHandle) {
				if (session->callBackHandle->onDisConnectionCallBack) {
					session->callBackHandle->onDisConnectionCallBack(session, session->interData);
				}
			}
			//shutdownSession(session);
			destroySessionEvent(session->t, session);
			removeSessionEvent(session);
		} else if (ret == CONTINUE_HANDLE) {
			if (getReadableBytesBuffer(session->readBuffer) <= 0) {
				retrieveAllBuffer(session->readBuffer);
			}
		} else if (ret == CLOSE_HANDLE) {
			if (session->callBackHandle) {
				if (session->callBackHandle->onDisConnectionCallBack) {
					session->callBackHandle->onDisConnectionCallBack(session, session->interData);
				}
			}
			//destroySession(session);
			destroySessionEvent(session->t, session);
			removeSessionEvent(session);
		}
	}
}

void handleFileReadEvent(struct session* session, struct thread* t) {
	mylog(LOG_INFO, "in handle File ReadEvent!!!");
	if (!session || !t ) {
		mylog(LOG_WARNING, "onMsgqueueEvent, session or thread is null!!!");
		return;
	}

	int fd = session->cfd;
	if (fd < 0) {
		mylog(LOG_WARNING, "onMsgqueueEvent some error, session id[%d], thread id[%d]", session->id, t->threadId);
		destroySession(session);
		session = NULL;
		return;
	}

	session->t = t;
	int ret = aeCreateFileEvent(t->threadEventLoop, fd, AE_READABLE, onFileReadEvent, session);
	if (ret == AE_ERR){
		mylog(LOG_WARNING, "onMsgqueueEvent some error, create File Event error!!! session id[%d], thread id[%d]", session->id, t->threadId);
		destroySession(session);
		session = NULL;
		return;
	}

	//dictAdd(t->sessionDict, (void*)session, (void*)session);
	pushBackFastArray(t->sessionArray, (struct arrayItem* )session);

	if (session->callBackHandle) {
		if (session->callBackHandle->onConnectionCallBack) {
			session->callBackHandle->onConnectionCallBack(session, session->interData);
		}
	}
}

typedef struct AsyncSendContent {
	struct session* session;
	char* sendBuf;
	int sendBufLen;
} AsyncSendContent;

void handleSendEvent(struct AsyncSendContent* content, struct thread* t) {
	mylog(LOG_INFO, "ONSENDEVENT!!!");
	if (!content) {
		return;
	}

	struct session* session = content->session;
	char* sendBuf = content->sendBuf;
	int sendBufLen = content->sendBufLen;
	writeSession(session, sendBuf, sendBufLen);
	if (sendBuf) {
		zfree(sendBuf);
	}
	zfree(content);

	// destroySession ??
	//if (session->t){
	//	if (session->t->threadState == STOPING_THREADSTATE) {
	//		mylog(LOG_WARNING, "%%%%%%%%%%%%%%5");
	//		destroySession(session);
	//	}
	//}
}

void handleStopThread(struct thread* t) {
	if (!t) {
		return;
	}
	mylog(LOG_DEBUG, "@@@@@@STOP THREAD [%d]\n\n", t->threadId);
	t->threadState = STOPING_THREADSTATE;
	if(lengthMsgqueue(t->threadEventQueue) > 0) {
		mylog(LOG_INFO, "@@@stop thread[%d]ing!!!", t->threadId);
		stopThreadEvent(t);
		mylog(LOG_DEBUG, "@@@@@@STOP THREAD [%d]", t->threadId);
		return;
	}

	mylog(LOG_DEBUG, "STOP THREAD [%d]", t->threadId);
	t->threadState =STOP_THREADSTATE;
}

void handleDestroySession(struct session* session) {
	if (!session) {
		return;
	}

	//dictDeleteNoFree(session->t->sessionDict, (void*)session);
	eraseFastArray(session->t->sessionArray, (struct arrayItem*)session);

	destroySession(session);
	session = NULL;
}

void onMsgqueueEvent(void* arg1, void* arg2) {
	struct msgQueueEvent* event = (struct msgQueueEvent*)arg1;
	struct thread* t = (struct thread* )arg2;
	if (!event || !t) {
		return;
	}

	mylog(LOG_INFO, "ON MSGQUEUEeVENT!!!!!");
	if (event->eventType == FILEREADEVENT_TYPE) {
		handleFileReadEvent((struct session*)event->privateData, t);
	} else if (event->eventType == SENDEVENT_TYPE) {
		handleSendEvent((struct AsyncSendContent*)event->privateData, t);
	} else if (event->eventType == STOPEVENT_TYPE) {
		handleStopThread(t);
	} else if (event->eventType == DESTROYSESSION_TYPE	) {
		handleDestroySession((struct session*) event->privateData);
	}

	zfree(event);
}

void* eventLoopThread(void* args) {
	struct thread* t = (thread*) args;
	if (!t) {
		return NULL;
	}

	aeEventLoop* el = t->threadEventLoop;
	if (!el) {
		return NULL;
	}

	t->threadState = START_THREADSTATE;
	mylog(LOG_INFO, "eventloop thread[%d] start!!!", t->threadId);
	aeMain(t->threadEventLoop);
	mylog(LOG_INFO, "eventloop thread[%d] end!!!", t->threadId);
	return NULL;
}

int threadCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
	struct thread* t = (struct thread*) clientData;
	if (!t) {
		return AE_NOMORE;
	}

	if (t->threadState == STOP_THREADSTATE) {
		mylog(LOG_INFO, "thread[%d] corn!!!", t->threadId);
		if (t->threadEventLoop) {
			aeStop(t->threadEventLoop);
			t->threadState = STOPING_THREADSTATE;
			return AE_NOMORE;
		}
	}
	return 100; /* every 100ms */
}

void stopThread(struct thread* t) {
	if (!t) {
		return;
	}

	mylog(LOG_INFO, "stop thread[%d]", t->threadId);
	t->threadState = STOP_THREADSTATE;
}

int dictEncObjKeyCompare(void *privdata, const void *key1, const void *key2)
{
	struct session* session1 = (struct session* )key1;
	struct session* session2 = (struct session* )key2;

    return session1->id == session1->id;
}

unsigned int dictEncObjHash(const void *key) {
	struct session* session = (struct session*)key;
	return session->id;
}

/* sessionDict type */
dictType sessionDictType = {
    dictEncObjHash,            /* hash function */
    NULL,                      /* key dup */
    NULL,                      /* val dup */
    dictEncObjKeyCompare,      /* key compare */
    NULL, /* key destructor */
    NULL                       /* val destructor */
};

#define MSGQUEUESIZE	1024
struct thread* initThread() {
	struct thread* t = NULL;
	if ( ! (t = (struct thread* )zcalloc(sizeof(struct thread)) ) ) {
		mylog(LOG_WARNING, "zcalloc thread error!!!");
		return NULL;
	}

	t->threadEventLoop = aeCreateEventLoop();
	if (! t->threadEventLoop) {
		mylog(LOG_WARNING, "in initThread create event loop error!!!");
		zfree(t);
		return NULL;
	}

	//t->sessionDict = dictCreate(&sessionDictType,NULL);
	//if (!t->sessionDict) {
	//	mylog(LOG_WARNING,  "in initThread create session dict error!!!");
	//	aeDeleteEventLoop(t->threadEventLoop);
	//	zfree(t);
	//	return NULL;
	//}
	t->sessionArray = initFastArray(1024);
	if (!t->sessionArray) {
		mylog(LOG_WARNING, "in initThread initFastArray error!!!");
		aeDeleteEventLoop(t->threadEventLoop);
		zfree(t);
		return NULL;
	}

	t->threadEventQueue = (struct eventMsgqueue*) initMsgqueue(t->threadEventLoop, MSGQUEUESIZE, onMsgqueueEvent, t);
	if (!t->threadEventQueue) {
		mylog(LOG_WARNING,  "in initThread init msg queue error!!!");
		aeDeleteEventLoop(t->threadEventLoop);
		zfree(t);
		return NULL;
	}

	t->threadId = ++sThreadId;

    //  Following code will guarantee more predictable latecnies as it'll
    //  disallow any signal handling in the I/O thread.
    //sigset_t signal_set;
    //int rc = sigfillset (&signal_set);
    //rc = pthread_sigmask (SIG_BLOCK, &signal_set, NULL);
	int ret = pthread_create( &t->threadPid, NULL, eventLoopThread, (void* )t );
	if (ret == 0) {
		mylog(LOG_INFO, "create thread[%d] success!!!", t->threadId);
	} else {
		mylog(LOG_WARNING, "in initThread create thread[%d] error!!!\n", t->threadId);
		destroyMsgqueue(t->threadEventQueue);
		aeDeleteEventLoop(t->threadEventLoop);
		zfree(t);
		return NULL;
	}

	aeCreateTimeEvent(t->threadEventLoop,
										1,
										threadCron,
										t,
										NULL);
	t->threadState = INIT_THREADSTATE;
	return t;
}

void destroyThread(struct thread* t) {
	if (!t) {
		mylog(LOG_WARNING, "destroy thread error, thread t is null!!!\n");
		return;
	}

	mylog(LOG_INFO, "destroy thread[%d]ing!!!", t->threadId);

	pthread_t tid = t->threadPid;
	pthread_join(tid, NULL);

	//if (t->sessionDict) {
	//	mylog(LOG_INFO, "DESTRORY===========");
  //      dictIterator *di;
   //     dictEntry *setele;
	//    di = dictGetIterator(t->sessionDict);
	//    while((setele = dictNext(di)) != NULL) {
	//    	struct session* session = NULL;
	//    	session = dictGetEntryVal(setele);
	//    	destroySession(session);
	//    }
	//    dictReleaseIterator(di);
	//	mylog(LOG_INFO, "DESTRORY===========");
	//	dictRelease(t->sessionDict);
	//}

	if (t->sessionArray) {
		int tail = getTailFastArray(t->sessionArray);
		while(tail > 0) {
			struct arrayItem* arrayItem = getArrayItemFastArray(t->sessionArray, tail - 1);
			if (!arrayItem) {
				mylog(LOG_INFO, "arrayItem break!!!");
				break;
			}
			eraseFastArray(t->sessionArray, arrayItem);
			destroySession((struct session*)arrayItem);
			tail = getTailFastArray(t->sessionArray);
		}

		destroyFastArray(t->sessionArray);
	}

	mylog(LOG_INFO, "thread[%d] exit!!!", t->threadId);
	destroyMsgqueue(t->threadEventQueue);
	aeDeleteEventLoop(t->threadEventLoop);
	zfree(t);
	mylog(LOG_INFO, "~~~~~~~~~~~~~~~~~~~~~~~~\n\n");
	return;
}

void sendMsgInThread(struct session* session, char* sendBuf, int sendBufLen) {
	if (!session) {
		return;
	}

	struct thread* t = session->t;
	if (t->threadState == STOPING_THREADSTATE) {
		mylog(LOG_WARNING, "thread id[%d] stoping!!!", t->threadId);
		// destroySession??
		destroySession(session);
		zfree(sendBuf);
		return;
	}

	struct eventMsgqueue* theadEventQueue = t->threadEventQueue;
	if (! theadEventQueue) {
		mylog(LOG_WARNING, "in sendMsgInThread, threadEventQueue is null!!!");
		return;
	}

	struct msgQueueEvent* msgQueueEvent = (struct msgQueueEvent* )zcalloc(sizeof(struct msgQueueEvent));
	if (!msgQueueEvent) {
		return;
	}

	struct AsyncSendContent* sendContent = (struct AsyncSendContent*)zcalloc(sizeof(struct AsyncSendContent));
	if (!sendContent) {
		mylog(LOG_WARNING, "in despathWorkThread, sendcontent is null!!!");
		zfree(msgQueueEvent);
		return;
	}
	sendContent->sendBuf = sendBuf;
	sendContent->sendBufLen = sendBufLen;
	sendContent->session = session;

	msgQueueEvent->eventType = SENDEVENT_TYPE;
	msgQueueEvent->privateData = sendContent;
	int ret = pushMsgqueue(theadEventQueue, msgQueueEvent);
	if (ret == -1) {
		mylog(LOG_WARNING, "***IN sendMsgInThread ERROR!!!");
		zfree(msgQueueEvent);
		zfree(sendContent);
		zfree(sendBuf);
	}
}

int pushMsgQueueEvent(struct thread* t, struct session* session) {
	if (!t || !session) {
		return -1;
	}

	if (t->threadState == STOPING_THREADSTATE) {
		mylog(LOG_WARNING, "thread id[%d] is stoping, don't push msg queue event!!!", t->threadId);
		return -1;
	}

	struct eventMsgqueue* threadEventQueue = t->threadEventQueue;
	if (! threadEventQueue) {
		mylog(LOG_WARNING, "in pushMsgQueueEvent, threadEventQueue is null!!!");
		return -1;
	}

	struct msgQueueEvent* msgQueueEvent = (struct msgQueueEvent* )zcalloc(sizeof(struct msgQueueEvent));
	if (!msgQueueEvent) {
		return -1;
	}

	msgQueueEvent->eventType = FILEREADEVENT_TYPE;
	msgQueueEvent->privateData = session;
	int ret = pushMsgqueue(threadEventQueue, msgQueueEvent);
	if (ret == -1) {
		mylog(LOG_WARNING, "***IN PUSH MSG QUEUEeVENT ERROR!!!");
		zfree(msgQueueEvent);
		destroySession(session);
		session = NULL;
	}
	return 0;
}

void stopThreadEvent(struct thread* t) {
	if (!t) {
		return;
	}

	struct eventMsgqueue* threadEventQueue = t->threadEventQueue;
	if (! threadEventQueue) {
		mylog(LOG_WARNING, "in pushMsgQueueEvent, threadEventQueue is null!!!");
		return;
	}

	struct msgQueueEvent* msgQueueEvent = (struct msgQueueEvent* )zcalloc(sizeof(struct msgQueueEvent));
	if (!msgQueueEvent) {
		return;
	}

	msgQueueEvent->eventType = STOPEVENT_TYPE;
	msgQueueEvent->privateData = NULL;
	int ret = pushMsgqueue(threadEventQueue, msgQueueEvent);
	if (ret == -1) {
		mylog(LOG_WARNING, "***IN stopThreadEvent ERROR!!!");
		zfree(msgQueueEvent);
	}
}

void destroySessionEvent(struct thread* t, struct session* session) {
	if (!t) {
		return;
	}

	struct eventMsgqueue* threadEventQueue = t->threadEventQueue;
	if (! threadEventQueue) {
		mylog(LOG_WARNING, "in pushMsgQueueEvent, threadEventQueue is null!!!");
		return;
	}

	struct msgQueueEvent* msgQueueEvent = (struct msgQueueEvent* )zcalloc(sizeof(struct msgQueueEvent));
	if (!msgQueueEvent) {
		return;
	}

	msgQueueEvent->eventType = DESTROYSESSION_TYPE;
	msgQueueEvent->privateData = session;
	int ret = pushMsgqueue(threadEventQueue, msgQueueEvent);
	if (ret == -1) {
		mylog(LOG_WARNING, "***IN destroySessionEvent ERROR!!!");
		zfree(msgQueueEvent);
		destroySession(session);
		session = NULL;
	}
}
